package net.bwie.vehicle.dws.job1;

import com.alibaba.fastjson.JSON;

import net.bwie.realtime.guanjuntao.util.DateTimeUtil;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import net.bwie.vehicle.dws.bean1.VehicleData;
import net.bwie.vehicle.dws.function1.AbnormalReportAnEmergencyFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * AbnormalReportAnEmergencyJob类用于处理车辆异常报告中的紧急情况
 */
public class AbnormalReportAnEmergencyJob {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1
        env.setParallelism(1);
        // 从Kafka中消费车辆数据
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "vehicle-data");
        // 处理车辆数据并返回报警流
        DataStream<String> alarmStream = processVehicleData(dataStream);
        alarmStream.print();
        //保存数据
        JdbcUtil.sinkToClickhouseUpsert(alarmStream, "insert into new_car.dws_Abonrmal (startTime, endTime,vin,alarm_type)\n" +
                "values (?, ?,?,?)");
        // 执行任务
        env.execute("VehicleAbnormalAlarm1Job");
    }

    /**
     * 处理车辆数据，识别异常高温报警模式
     *
     * @param dataStream 输入的车辆数据流
     * @return 返回识别到的报警信息流
     */
    private static DataStream<String> processVehicleData(DataStream<String> dataStream) {
        // 将数据流中的数据转换为VehicleData对象
        SingleOutputStreamOperator<VehicleData> map = dataStream.map(o -> JSON.parseObject(o, VehicleData.class));
        // 为数据流添加水印
        SingleOutputStreamOperator<VehicleData> watermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<VehicleData>() {
                    @Override
                    public long extractTimestamp(VehicleData element, long recordTimestamp) {
                        return element.getTimestamp();
                    }
                })
        );
        // 按照车辆标识符进行分组
        KeyedStream<VehicleData, String> vehicleDataStringKeyedStream = watermarks.keyBy(VehicleData::getVin);
        // 定义电池最高温度阈值
        final Double MaxBatteryTemp = 60.0;
        // 定义匹配模式，用于检测连续三次电池温度超过阈值的事件
        Pattern<VehicleData, VehicleData> pattern = Pattern.<VehicleData>begin("a")
                // 定义一个条件，筛选出电池温度超过最大值的车辆数据
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) throws Exception {
                        // 检查电池温度是否非空且超过最大值
                        return value.getBatteryTemp() != null
                                && value.getBatteryTemp() > MaxBatteryTemp;
                    }
                }).times(3)
                .within(Time.seconds(30));

        // 创建模式流
        PatternStream<VehicleData> patternStream = CEP.pattern(vehicleDataStringKeyedStream, pattern);
        // 处理匹配到的模式并生成报警信息
        SingleOutputStreamOperator<String> restStream = patternStream.process(new PatternProcessFunction<VehicleData, String>() {
            @Override
            public void processMatch(Map<String, List<VehicleData>> match, Context ctx, Collector<String> out) throws Exception {
                // 获取标记为"a"的车辆数据事件列表
                List<VehicleData> events = match.get("a");
                // 如果事件列表为空或包含的事件少于3个，则不执行后续操作并退出
                if (events == null || events.size() < 3) return;
                // 获取事件列表中的第一个车辆数据
                VehicleData first = events.get(0);
                // 获取事件列表中的最后一个车辆数据
                VehicleData last = events.get(events.size() - 1);
                // 将第一个车辆数据的时间戳转换为可读的开始时间字符串
                String startTime = DateTimeUtil.convertLongToString(first.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);
                 // 将最后一个车辆数据的时间戳转换为可读的结束时间字符串
                String endTime = DateTimeUtil.convertLongToString(last.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);
                // 构建包含开始时间、结束时间和车辆识别号(VIN)的JSON字符串
                String json = String.format("{\"startTime\":\"%s\",\"endTime\":\"%s\",\"vin\":\"%s\",\"alarm_type\":\"连续三次高温报警\"}",
                        startTime, endTime, first.getVin());
                // 输出构建的JSON字符串
                out.collect(json);


            }
        });
        // 处理车辆数据中的异常报告和紧急情况
        SingleOutputStreamOperator<String> processStream = vehicleDataStringKeyedStream.process(new AbnormalReportAnEmergencyFunction());
        // 合并报警信息流和处理流，并返回
        return processStream.union(restStream);
    }
}
